package com.techsun.sink;

import com.alibaba.fastjson.JSONObject;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.instrumentation.SinkCounter;
import org.apache.flume.sink.AbstractSink;
import org.apache.log4j.Logger;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.LinkedList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * metric 各参数含义
 * BatchCompleteCount(完成的批数量)、
 * ConnectionFailedCount（连接失败数）、
 * EventDrainAttemptCount（尝试提交的event数量）、
 * ConnectionCreatedCount（创建连接数）、
 * Type（组件类型）、
 * BatchEmptyCount（批量取空的数量）、
 * ConnectionClosedCount（关闭连接数量）、
 * EventDrainSuccessCount（成功发送event的数量）、
 * StartTime（组件开始时间）、
 * StopTime（组件停止时间）、
 * BatchUnderflowCount（正处于批量处理的batch数）
 * <p>
 * channel
 * EventPutSuccessCount（成功放入channel的event数量）
 * ChannelFillPercentage（通道使用比例）、Type（组件类型）
 * EventPutAttemptCount（尝试放入将event放入channel的次数）
 * ChannelSize（目前在channel中的event数量）
 * StartTime（组件开始时间）
 * StopTime（组件停止时间）
 * EventTakeSuccessCount（从channel中成功取走的event数量）
 * ChannelCapacity（通道容量）
 * EventTakeAttemptCount（尝试从channel中取走event的次数）
 */
public class ShellSink extends AbstractSink implements Configurable {
    Logger logger = Logger.getLogger(ShellSink.class);
    private Long takeEventIntervalSecond;
    private SinkCounter sinkCounter;
    private String scriptCommand;
    // 正则匹配到的字符串
    private List<String> scriptCommandMatchList = new LinkedList<>();

    @Override
    public void configure(Context context) {
        this.scriptCommand = context.getString("scriptCommand");
        this.takeEventIntervalSecond = context.getLong("takeEventIntervalSecond", 60L);
        // 使用正则匹配
        Matcher m = Pattern.compile("\\$\\{([a-z]+)\\}").matcher(scriptCommand);
        while (m.find()) {
            this.scriptCommandMatchList.add(m.group(1));
        }
        if (this.sinkCounter == null) {
            this.sinkCounter = new SinkCounter(this.getName());
        }
    }


    private void execShellCommand(JSONObject jsonobject) throws Throwable {
        String commandStr = scriptCommand;
        for (String matchStr : this.scriptCommandMatchList) {
            commandStr = commandStr.replace("${" + matchStr + "}", jsonobject.getString(matchStr));
        }
        String[] commandArray = commandStr.split(" ");
        logger.info(commandStr);
        Process process = Runtime.getRuntime().exec(commandArray);
        int exitValue = process.waitFor();
        if (0 != exitValue) {
            BufferedReader input = new BufferedReader(new InputStreamReader(process.getErrorStream()));
            StringBuilder result = new StringBuilder();
            String line = "";
            while ((line = input.readLine()) != null) {
                result.append(line);
            }
            input.close();
            String error = String.format("call script command: %s failed, exitStatus: [%s], result: [%s]", commandStr, exitValue, result);
            throw new Throwable(error);
        }
        process.destroy();
    }

    @Override
    public Status process() throws EventDeliveryException {
        //声明返回值状态信息
        Status status = null;

        //获取当前sink绑定的channel
        Channel channel = getChannel();

        //获取事务
        Transaction transaction = channel.getTransaction();


        try {
            //开启事务
            transaction.begin();
            //声明事件
            Event event;
            while ((event = channel.take()) == null) {
                this.sinkCounter.incrementBatchEmptyCount();
                Thread.sleep(this.takeEventIntervalSecond * 1000);
            }
            JSONObject outJson = JSONObject.parseObject(new String(event.getBody()));
            this.sinkCounter.incrementEventDrainAttemptCount();
            this.execShellCommand(outJson);
            //事务提交
            transaction.commit();
            this.sinkCounter.incrementEventDrainSuccessCount();
            status = Status.READY;
        } catch (Throwable e) {
            //当遇到异常的时候，回滚事务
            transaction.rollback();
            status = Status.BACKOFF;
            throw new EventDeliveryException(e);
        } finally {
            //关闭事务
            transaction.close();
        }

        return status;
    }

    @Override
    public synchronized void start() {
        this.sinkCounter.start();
        super.start();
    }

    @Override
    public synchronized void stop() {
        this.sinkCounter.stop();
        super.stop();
    }
}
